﻿#region LicenseHeader

// Copyright 2012 The Trustees of Indiana University.  All rights reserved.
// 
// The Indiana University licenses this file to you under the Apache License, 
// Version 2.0 (the "License"); you may not use this file except in compliance 
// with the License.  You may obtain a copy of the License at 
// 
// http://www.apache.org/licenses/LICENSE-2.0
// 
// Unless required by applicable law or agreed to in writing, software 
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
//  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  
// See the License for the specific language governing permissions and 
// limitations under the License.
// 
// @author Thilina Gunarathne (tgunarat@indiana.edu)

#endregion

/*using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using AzureMRCore.DataModel;
using AzureMRCore.DataTypes;
using AzureMRCore.InputFormat;
using Microsoft.WindowsAzure;
using Microsoft.WindowsAzure.StorageClient;

namespace AzureMRCore.Primitives
{
    public class BroadcastReceiverDeamon
    {
        private readonly ConcurrentDictionary<string, IInputFormat> _inMemCache;
        private readonly ConcurrentDictionary<string, Task> _fetchingBcastData;

        private int _pollingInterval = 100;
        private string _tableAddress;
        private StorageCredentials _storageCredentials;
        private CloudTableClient _tableClient;
        private CloudStorageAccount _storageAccount;
        private ErrorDataContext _errorContext;

        public BroadcastReceiverDeamon(CloudStorageAccount storageAccount)
        {
            _tableAddress = storageAccount.TableEndpoint.ToString();
            _storageCredentials = storageAccount.Credentials;
            _storageAccount = storageAccount;
            _tableClient = _storageAccount.CreateCloudTableClient();
            _inMemCache = new ConcurrentDictionary<string, IInputFormat>();
            _fetchingBcastData = new ConcurrentDictionary<string, Task>();
            _errorContext = new ErrorDataContext(_tableAddress,
                                                 _storageCredentials, "AzureMRErrors");
            _tableClient.CreateTableIfNotExist(_errorContext.Name);
        }

        public void RunBroadcastReceiver()
        {
            var bcastDataContext = new BroadcastDataContext(_tableAddress, _storageCredentials,
                                                            DataModelUtils.GetBroadcastTableName());
            _tableClient.CreateTableIfNotExist(bcastDataContext.Name);

            while (true)
            {
                try
                {
                    /* First check the task table#1#
                    var bcastEnum = bcastDataContext.Tasks;
                    var broadcastDataModels = bcastEnum.ToList();
                    foreach (var bCastData in broadcastDataModels)
                    {
                        if (!_inMemCache.ContainsKey(bCastData.BCastDataTable) &&
                            !(_fetchingBcastData.ContainsKey(bCastData.BCastDataTable)))
                        {
                            BroadcastDataModel data = bCastData;
                            Task fetchTask = new Task(() => FetchBCastTableData(Int32.Parse(data.RowKey),
                                                                                data.JobID, data.AppName,
                                                                                data.NumMapTasks, data.BCastDataTable,
                                                                                data.BCastInputFormat));
                            _fetchingBcastData.TryAdd(bCastData.BCastDataTable, fetchTask);
                            fetchTask.Start();
                        }
                    }
                    foreach (KeyValuePair<string, IInputFormat> pair in _inMemCache)
                    {
                        string key = pair.Key;
                        var dataModels = broadcastDataModels.Count(record => (record.BCastDataTable == key));
                        if (dataModels == 0)
                        {
                            IInputFormat temp;
                            _inMemCache.TryRemove(key, out temp);
                        }
                    }
                }
                catch (Exception e)
                {
                    var error = new ErrorDataModel(e, "maptask", e.Message);
                    _errorContext.AddObject(_errorContext.Name, error);
                    _errorContext.SaveChangesWithRetries();
                }
                Thread.Sleep(_pollingInterval*5);
            }
        }

        public bool ContainsBCastData(string tableName)
        {
            return _inMemCache.ContainsKey(tableName);
        }

        public IInputFormat GetBCastData(string tableName)
        {
            var item = _inMemCache[tableName];
            return item;
        }

        public void FetchBCastTableData(int iteration, string jobID, string appName,
                                        int numMapTasks, string tableName, string inputFormatTypeString)
        {
            int retry = 0;
            bool isAllMapDone = false;
            int count = 0;
            var fetchedBCastData = new ConcurrentDictionary<int, BCastDataProduct>();
            var fetchedRecords = new ConcurrentDictionary<string, string>();

            var mapContext = new MapMetaDataContext(_tableAddress, _storageCredentials,
                                                    DataModelUtils.GetMapTableName(jobID));
            var reduceInputDataContext = new ReduceInputDataContext(_tableAddress, _storageCredentials,
                                                                    tableName);
            Type inputFormatType = Type.GetType(inputFormatTypeString);
            var inputFormat = (IInputFormat) Activator.CreateInstance(inputFormatType);

            reduceTaskDataFetchLabel:
            //Rather than failing, do some retries here 
            try
            {
                while (true)
                {
                    //right aligns the iteration number so that we can query the table based on partitions
//                    string iterationString = iteration.ToString().PadLeft(5, '0');
//                    string nextIter = (iteration + 1).ToString().PadLeft(5, '0');
                    //Querying the table based on partitions
                    IQueryable<ReduceInputDataModel> reduceTaskRecords = reduceInputDataContext.ReduceTaskRecords;

                    foreach (ReduceInputDataModel record in reduceTaskRecords)
                    {
                        if (record.Value == null | fetchedRecords.ContainsKey(record.RowKey)) continue;
                        string[] ids = record.MapID.Split('_');
                        int mapIndex = Int32.Parse(ids[0]);
                        bool isAdded = false;
                        isAdded = fetchedBCastData.TryAdd(mapIndex,
                                                          new BCastDataProduct()
                                                              {
                                                                  IsSingle = (record.ValueCount == 1),
                                                                  Key = record.Key,
                                                                  Value = inputFormat.FromBytes(record.Value)
                                                              });
//                        Trace.WriteLine("fetched " + record.PartitionKey + "  " + record.MapID);

                        if (isAdded)
                        {
                            fetchedRecords.TryAdd(record.RowKey, "");
                        }
                        count = 0;
                    }

                    //IsDoneFetching performs sleeping too. Exponential back-off using count*sleep
                    if (IsDoneFetching(numMapTasks, iteration.ToString(), appName, _pollingInterval*count,
                                       fetchedRecords.Count, mapContext, ref isAllMapDone))
                    {
//                        Trace.WriteLine("Leaving fetch..." + RoleEnvironment.CurrentRoleInstance.Id);
                        //combine the fetched records using format data in the table..
                        // is it possible to use something like a array combiner to avoid copying or diretly download to such a thing
                        inputFormat.Init(fetchedBCastData, numMapTasks);
                        _inMemCache.TryAdd(tableName, inputFormat);
                        Task task;
                        _fetchingBcastData.TryRemove(tableName, out task);
                        break;
                    }
                    count++;
                    retry = 0;
                }
            }
            catch (Exception e)
            {
                Trace.WriteLine(e.Message + "    " + e.StackTrace);
                if (retry < Constants.NUM_RETRIES)
                {
                    retry++;
                    Thread.Sleep(100);
                    goto reduceTaskDataFetchLabel;
                }
                var error = new ErrorDataModel(e, "broadcastfetch", e.Message);
                _errorContext.AddObject(_errorContext.Name, error);
                _errorContext.SaveChangesWithRetries();
            }
        }

        internal bool IsDoneFetching(int numMapTasks, string iteration, string appName, int pollingInterval,
                                     int fetchedBLOBCount, MapMetaDataContext mapContext, ref bool isAllMapDone)
        {
            if (!isAllMapDone)
            {
                IQueryable<MapDataModel> executingMapTasks =
                    mapContext.MapTasks.Where(
                        p => p.TaskStatus != MapDataModel.FINISHED & p.PartitionKey == iteration & p.AppName == appName)
                        .Take(1);

                int executingCount = Enumerable.Count(executingMapTasks);
                if (executingCount > 0)
                {
                    Thread.Sleep(pollingInterval);
                    return false;
                }

                //hnadling the case where the MapTasks are still not populated in to the table
                var allMapTasks =
                    mapContext.MapTasks.Where(p => p.PartitionKey == iteration & p.AppName == appName).
                        AsTableServiceQuery();
                int count = Enumerable.Count(allMapTasks);
                if (count < numMapTasks)
                {
                    Thread.Sleep(pollingInterval);
                    return false;
                }

                IQueryable<MapDataModel> doubleCheckMapTasks =
                    allMapTasks.Where(p => p.TaskStatus != MapDataModel.FINISHED).Take(1);
                //Done processing all the map tasks)
                if (Enumerable.Count(doubleCheckMapTasks) > 0)
                {
                    Thread.Sleep(pollingInterval);
                    return false;
                }
                isAllMapDone = true;
            }

            IQueryable<MapDataModel> mapTasks =
                mapContext.MapTasks.Where(p => p.PartitionKey == iteration & p.AppName == appName).AsTableServiceQuery();
            int totalDataProducts = Enumerable.Aggregate(mapTasks, 0,
                                                         (current, mapTask) =>
                                                         current + mapTask.ReduceProductCounts[0]);
            return fetchedBLOBCount == totalDataProducts;
        }
    }
}*/
